package tss.hbase;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.joda.time.Instant;

import tss.datastore.ValueType;
import tss.datastore.data.DataSeriesIdentifier;
import tss.datastore.data.InstantValue;

public class HBaseDataPersister {

    private static String dataTableName = "data";
    private static byte[] defaultFamily = Bytes.toBytes("cf");
    private static byte[] valueColumnName = Bytes.toBytes("ivs");
    private HBaseDriver hBaseDriver;

    public HBaseDataPersister(HBaseDriver hBaseDriver) {
        this.hBaseDriver = hBaseDriver;
        createDataTable();
    }
    
    private void createDataTable() {
        hBaseDriver.doWithAdminConnection(new HBaseAdminAction() {

            @Override
            public void perform(HBaseAdmin admin) throws IOException {
                if (!admin.tableExists(dataTableName)) {
                    HTableDescriptor hTableDescriptor = new HTableDescriptor(dataTableName);
                    hTableDescriptor.addFamily(new HColumnDescriptor(defaultFamily));
                    admin.createTable(hTableDescriptor);
                }
            }
        });
    }

    public void addDataPoint(final DataSeriesIdentifier seriesIdentifier, final Instant instant, final Object value,
            final ValueType type) {
        hBaseDriver.doWithTable(dataTableName, new HBaseTableAction() {

            @Override
            public void perform(HTable hTable) throws IOException {
                byte[] key = seriesIdentifier.toBytes();
                byte[] existingValueInBytes = loadExistingValue(hTable, key);

                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                DataOutputStream dos = new DataOutputStream(bos);
                dos.write(existingValueInBytes);
                appendInstantValue(instant, value, type, dos);
                dos.flush();

                byte[] seriesInBytes = bos.toByteArray();
                Put put = new Put(key);
                put.add(defaultFamily, valueColumnName, seriesInBytes);
                hTable.put(put);
            }

            private void appendInstantValue(Instant instant, Object value, ValueType type, DataOutputStream os)
                    throws IOException {
                long millis = instant.getMillis();
                os.writeLong(millis);

                switch (type) {
                case INT:
                    os.writeInt((int) value);
                    break;
                case STRING:
                    os.writeUTF((String) value);
                    break;
                default:
                    throw new RuntimeException("Value type " + type + " is not supported.");
                }
            }
        });

    }

    public <T> List<InstantValue<T>> getDataSeries(final DataSeriesIdentifier seriesIdentifier, final ValueType type) {
        return hBaseDriver.queryTable(dataTableName, new HBaseTableQuery<List<InstantValue<T>>>() {
            @Override
            public List<InstantValue<T>> perform(HTable hTable) throws IOException {
                byte[] key = seriesIdentifier.toBytes();
                byte[] existingValueInBytes = loadExistingValue(hTable, key);

                ByteArrayInputStream bis = new ByteArrayInputStream(existingValueInBytes);
                DataInputStream dis = new DataInputStream(bis);
                ArrayList<InstantValue<T>> result = new ArrayList<InstantValue<T>>();
                InstantValue<T> instantValue = null;
                while ((instantValue = retrieveInstantValue(dis, type)) != null) {
                    result.add(instantValue);
                }
                return result;
            }

            private InstantValue<T> retrieveInstantValue(DataInputStream dis, ValueType type) throws IOException {
                try {
                    long millis = dis.readLong();
                    Instant instant = new Instant(millis);
                    Object value;
                    switch (type) {
                    case INT:
                        value = dis.readInt();
                        break;
                    case STRING:
                        value = dis.readUTF();
                        break;
                    default:
                        throw new RuntimeException("Value type " + type
                                + " is not supported.");
                    }
                    return new InstantValue<T>(instant, value);
                } catch (EOFException e) {
                    return null;
                }
            }
        });
    }

    private byte[] loadExistingValue(HTable hTable, byte[] key) throws IOException {
        Get get = new Get(key);
        get.addColumn(defaultFamily, valueColumnName);
        Result result = hTable.get(get);
        if (result.isEmpty()) {
            return new byte[0];
        } else {
            return result.getValue(defaultFamily, valueColumnName);
        }
    }

    public void cleanAll() {
        hBaseDriver.doWithAdminConnection(new HBaseAdminAction() {

            @Override
            public void perform(HBaseAdmin admin) throws IOException {
                if (admin.tableExists(dataTableName)) {
                    admin.disableTable(dataTableName);
                    admin.deleteTable(dataTableName);
                }
            }
        });
        createDataTable();        
    }

}
